package cn.doitedu.dwetl

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.SparkSession

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-01-12
 * @desc GEOHASH地理位置字典生成器
 */
object GeohashDictGen {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("geohash字典生成")
      .master("local[*]")
      .getOrCreate()


    // 读取mysql服务器中的原始gps坐标数据
    val props = new Properties()
    props.setProperty("user","root")
    props.setProperty("password","ABC123abc.123")
    val frame = spark.read.jdbc("jdbc:mysql://hdp01.doitedu.cn:3306/realtimedw", "area_dict", props)

    frame.createTempView("df")

    // 将gps坐标转成geohash码
    val gps2Geohash = (lat:Double,lng:Double)=>{
      GeoHash.geoHashStringWithCharacterPrecision(lat,lng,5)
    }
    spark.udf.register("gps2geohash",gps2Geohash)
    val res = spark.sql(
      """
        |
        |select
        |  gps2geohash(BD09_LAT,BD09_LNG) as geohash,
        |  province,
        |  city,
        |  region
        |from df
        |
        |""".stripMargin)


    // 将结果以parquet文件格式，写入hdfs
    res.write.parquet("hdfs://hdp01.doitedu.cn:8020/dicts/geodict/")

    // hallo

    spark.close()
  }

}
